package com.winit.schedule.job.tracker.monitor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.event.EventHandler;
import com.winit.schedule.core.repository.constants.TaskItemStatus;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.repository.query.TaskItemQuery;
import com.winit.schedule.core.service.AbstractService;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.job.tracker.JobTrackerContextImpl;
import com.winit.schedule.job.tracker.conf.JobConfiguration;
import com.winit.schedule.job.tracker.event.ClusterEvent;
import com.winit.schedule.job.tracker.event.ClusterEventType;
import com.winit.schedule.job.tracker.event.TaskTrackerChangeEvent;
import com.winit.schedule.job.tracker.event.TaskTrackerEventType;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;

/**
 * tasktracker是否断连监控类
 * 
 * @author jianke.zhang 2015-3-21 下午4:08:08
 * @since 1.0
 */
public class TaskTrackerMonitor extends AbstractService implements EventHandler<ClusterEvent> {

    private static transient final Logger  logger         = LoggerFactory.getLogger(TaskTrackerMonitor.class);

    // 用来定时发送心跳
    private final ScheduledExecutorService taskExecutor   = Executors.newScheduledThreadPool(3);

    private long                           heartbeatTimeout;

    private long                           heartbeatCheckInterval;

    private JobTrackerContextImpl          jobTrackerContext;

    private ScheduleDataManager            scheduleDataManager;

    private volatile boolean               isStartMonitor = false;

    public TaskTrackerMonitor(JobTrackerContextImpl jobTrackerContext){
        this("TaskTrackerMonitor", jobTrackerContext);
        this.scheduleDataManager = jobTrackerContext.getScheduleDataManager();
    }

    public TaskTrackerMonitor(String name, JobTrackerContextImpl jobTrackerContext){
        super(name);
        this.jobTrackerContext = jobTrackerContext;
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);
        heartbeatTimeout = conf.getLong(JobConfiguration.HEARTBEAT_TIMEOUT_MS,
            JobConfiguration.DEFAULT_HEARTBEAT_TIMEOUT_MS);
        heartbeatCheckInterval = conf.getLong(JobConfiguration.HEARTBEAT_CHECK_INTERVAL_MS,
            JobConfiguration.DEFAULT_HEARTBEAT_CHECK_INTERVAL_MS);
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        taskExecutor.scheduleWithFixedDelay(new disconnentThread(this),
            2000,
            heartbeatCheckInterval,
            TimeUnit.MILLISECONDS);
        taskExecutor.scheduleWithFixedDelay(new ClearThread(this), 2000, heartbeatTimeout / 3, TimeUnit.MILLISECONDS);
        taskExecutor.scheduleWithFixedDelay(new CheckTimeoutTaskItem(this), 2000, 10 * 60 * 1000, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
        taskExecutor.shutdown();
    }

    public JobTrackerContextImpl getJobTrackerContext() {
        return jobTrackerContext;
    }

    public void setJobTrackerContext(JobTrackerContextImpl jobTrackerContext) {
        this.jobTrackerContext = jobTrackerContext;
    }

    public long getHeartbeatTimeout() {
        return heartbeatTimeout;
    }

    /**
     * 定时检测push至tasktracker失败的任务,并将推送失败的任务项状态重置为INACTIVITE,执行间隔为一个心跳超时时间执行3次
     */
    class ClearThread implements Runnable {

        TaskTrackerMonitor taskTrackerMonitor;

        public ClearThread(TaskTrackerMonitor taskTrackerMonitor){
            this.taskTrackerMonitor = taskTrackerMonitor;
        }

        @Override
        public void run() {
            if (isStartMonitor) {
                try {
                    Thread.currentThread().setPriority(Thread.MAX_PRIORITY);

                    // if (logger.isDebugEnabled()) {
                    // logger.debug("-----------recovery push failure taskitem------------");
                    // }

                    scheduleDataManager.recoveryPushFailureTaskItem(heartbeatTimeout);

                    // scheduleDataManager.cleanQuartzTimeout(heartbeatTimeout);

                } catch (Exception ex) {
                    logger.error("Detection push failed mission fails", ex);
                }
            }

        }
    }

    /**
     * 检查tasktracker心跳周期是否断开连接, 如果一个心跳周期内断开连接，清除缓存和数据库中的节点信息
     * 
     * @author jianke.zhang 2015-3-21 下午3:38:38
     * @since 1.0
     */
    class disconnentThread implements Runnable {

        TaskTrackerMonitor taskTrackerMonitor;

        long               startupTime = 0;

        int                count = 0;

        public disconnentThread(TaskTrackerMonitor taskTrackerMonitor){
            this.taskTrackerMonitor = taskTrackerMonitor;
            this.startupTime = System.currentTimeMillis();
        }

        @SuppressWarnings("unchecked")
        @Override
        public void run() {
            if (isStartMonitor) {
                Thread.currentThread().setPriority(Thread.MAX_PRIORITY);

                // job tracker启动后一个超时周期内检测残留的数据库tasktracker信息
                if (System.currentTimeMillis() - startupTime > heartbeatTimeout && count++ == 0) {
                    try {
                        logger.info("===> Cleaning the lose efficacy tasktracker");

                        scheduleDataManager.removeTaskTrackerAndRunnerForTimeout(heartbeatTimeout);
                    } catch (Exception ex) {
                        logger.error("Cleaning the lose efficacy tasktracker failure.", ex);
                    }
                }

                try {
                    // 检测task tracker断开连接
                    long currentTime = Clock.getSystemTime();
                    TaskTrackerEntity[] taskTrackers = jobTrackerContext.getTaskTrackers()
                        .values()
                        .toArray(new TaskTrackerEntity[] {});

                    List<String> disconnects = new ArrayList<String>();
                    for (TaskTrackerEntity taskTracker : taskTrackers) {
                        if (taskTracker.getHeartbeatTime() + heartbeatTimeout <= currentTime) {
                            logger.info("Task Tracker disconnect. uuid:" + taskTracker.getTaskTrackerUuid());

                            jobTrackerContext.unregisterTaskTracker(taskTracker);

                            scheduleDataManager.removeTaskTrackerAndRunnerByUuid(taskTracker.getTaskTrackerUuid());

                            disconnects.add(taskTracker.getTaskTrackerUuid());
                        }
                    }

                    if (!disconnects.isEmpty()) {
                        // 异步事件
                        jobTrackerContext.getDispatcher()
                                .getEventHandler()
                                .handle(new TaskTrackerChangeEvent(TaskTrackerEventType.DISCONNECT, disconnects));
                    }
                } catch (Exception ex) {
                    logger.error("Check TaskTracker disconnect failure.", ex);
                }
            }
        }
    }

    class CheckTimeoutTaskItem extends Thread {

        TaskTrackerMonitor taskTrackerMonitor;

        public CheckTimeoutTaskItem(TaskTrackerMonitor taskTrackerMonitor){
            this.taskTrackerMonitor = taskTrackerMonitor;
        }

        @SuppressWarnings("unchecked")
        @Override
        public void run() {
            if (isStartMonitor) {
                try {

                    Configuration conf = jobTrackerContext.getConf();
                    long timeout = conf.getLong(JobConfiguration.HEARTBEAT_TIMEOUT_MS,
                        JobConfiguration.DEFAULT_HEARTBEAT_TIMEOUT_MS);
                    TaskItemQuery query = new TaskItemQuery();
                    query.setTimeout(Clock.getSystemTime() - timeout);
                    query.setTaskItemStatusList(TaskItemStatus.RUN_STATUS);
                    List<TaskItemEntity> items = scheduleDataManager.queryTaskItemsByTimeout(query);

                    if (null != items && items.size() > 0) {
                        StringBuilder taskItems = new StringBuilder();
                        for (TaskItemEntity item : items) {
                            taskItems.append(item.getTaskItemId() + "\n\r");
                        }
                        jobTrackerContext.getDispatcher()
                            .getEventHandler()
                            .handle(new TaskTrackerChangeEvent(TaskTrackerEventType.TIMEOUT_TASKITEM, taskItems.toString()));
                    }

                } catch (Exception e) {
                    logger.error("CheckTimeoutTaskItem failure.", e);
                }
            }
        }
    }

    @Override
    public void handle(ClusterEvent event) {
        if (null != event && ClusterEventType.MASTER_CHANAE.equals(event.getType())) {
            if (event.isMaster()) {
                isStartMonitor = true;
            } else {
                isStartMonitor = false;
            }
        }
    }

}
